package com.liuqi.iot.service.flink;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.liuqi.iot.bean.dto.DeviceModelFieldDTO;
import com.liuqi.iot.data.AlarmCheckService;
import com.liuqi.iot.service.DeviceDataEntityService;
import com.liuqi.iot.service.DeviceEntityService;
import com.liuqi.iot.service.DeviceHistoryDataService;
import com.liuqi.iot.service.DeviceModelPublishedEntityService;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Properties;

/**
 * Flink原始工况加工处理
 *
 * @author  LiuQi 2025/3/26-10:41
 * @version V1.0
 **/
@Slf4j
@Service
@ConditionalOnProperty(prefix = "iot.flink", name = "enabled", havingValue = "true")
public class FlinkOriginDataProcessor {
    @Autowired
    private DeviceModelPublishedEntityService deviceModelPublishedEntityService;

    @Autowired
    private DeviceEntityService deviceService;

    @Autowired
    private DeviceDataEntityService deviceDataEntityService;

    @Autowired
    private DeviceHistoryDataService historyDataService;

    @Autowired
    private AlarmCheckService alarmCheckService;

    @Value("${iot.kafka.topic.processed}")
    private String processedTopic;

    @Value("${iot.kafka.servers}")
    private String servers;

    @Value("${iot.kafka.topic.origin}")
    private String originTopic;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void start() {
        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Kafka 连接属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", servers);
        properties.setProperty("group.id", "IOT_PLATFORM_FLINK");

        // 创建 Kafka 消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(originTopic, new SimpleStringSchema(), properties);

        env.addSource(consumer)
                .map(JSON::parseObject)
                .filter(obj -> {
                    String deviceCode = obj.getString("__deviceCode");
                    // 补充设备字段信息
                    // 处理属性
                    Map<String, DeviceModelFieldDTO> fieldMap = deviceModelPublishedEntityService.getDeviceFields(deviceCode);
                    if (null == fieldMap) {
                        return false;
                    }

                    obj.put("_fields", fieldMap);

                    return true;
                }).keyBy((KeySelector<JSONObject, String>) jsonObject -> jsonObject.getString("__deviceCode"))
                .process(new FlinkWindowProcessFunction())
//                .addSink()
                ;



        // 添加 Kafka 数据源
//        DataStream<String> stream = env.addSource(consumer);

//        stream.filter(StringUtils::isNotBlank)
//                .map(JSON::parseObject)
//                .filter(obj -> {
//                    String deviceCode = obj.getString("__deviceCode");
//                    // 补充设备字段信息
//                    // 处理属性
//                    Map<String, DeviceModelFieldDTO> fieldMap = deviceModelPublishedEntityService.getDeviceFields(deviceCode);
//                    if (null == fieldMap) {
//                        return false;
//                    }
//
//                    obj.put("_fields", fieldMap);
//
//                    return true;
//                })
//                .keyBy((KeySelector<JSONObject, Object>) jsonObject -> jsonObject.getString("__deviceCode"))
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
//                ;
//
//        // 执行 Flink 作业
//        try {
//            env.execute("Spring Flink Kafka Example");
//        } catch (Exception e) {
//            throw new RuntimeException(e);
//        }
    }
}
